package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object DataSourceAPIDemo {
  def main(args: Array[String]): Unit = {
    val ss: SparkSession = SparkSession.builder()
      .config("spark.sql.shuffle.partitions", "1") // 将产生的shuffle的分区设置为1
      .master("local")
      .appName("spark source api")
      .getOrCreate()

    import ss.implicits._
    import org.apache.spark.sql.functions._

    /**
     * ========================================读写csv文件===========================================
     */
    //读取csv文件数据
    //    val studentDF: DataFrame = ss.read
    //      .format("csv")
    //      .option("sep", ",")
    //      .schema("id STRING,name STRING,age INT,gender STRING,clazz STRING")
    //      .load("spark/data/students.txt")
    //
    ////    val studentDF: DataFrame = ss.read.csv("spark/data/students.txt")
    //
    //    val resRDD1: DataFrame = studentDF.groupBy("clazz")
    //      .agg(count("clazz") as "counts")
    //
    //    //写到csv文件中
    //    resRDD1.write
    //      .format("csv")
    //      .mode(SaveMode.Overwrite)
    //      .option("sep","|")
    //      .save("spark/output/out5")
    ////    resRDD1.write.csv("spark/output/out6")

    /**
     * ========================================读写json文件==========================================
     */
    //读取json数据文件
    //    val studentDF: DataFrame = ss.read
    //      .format("json")
    //      .load("spark/data/students.json")
    //
    //    val resRDD1: DataFrame = studentDF.groupBy("clazz")
    //      .agg(count("clazz") as "counts")
    //
    //    //写入到json文件中
    //    resRDD1.write.json("spark/output/out6")

    /**
     * ========================================读写parquet文件==========================================
     *
     * parquet格式文件的大小取决于原文件的【信息熵】
     * 将来文件中大量的数据都是相同的，可以采用parquet存储，占用更小的空间
     */

    //    val studentDF: DataFrame = ss.read
    //      .format("json")
    //      .load("spark/data/students2.json")
    //    //写入一个parquet格式的文件中
    //    studentDF.write.parquet("spark/output/out8")
    //
    //    val studentDF2: DataFrame = ss.read
    //      .format("json")
    //      .load("spark/data/students3.json")
    //    //写入一个parquet格式的文件中
    //    studentDF2.write.parquet("spark/output/out9")

    //    val resRDD1: DataFrame = ss.read.parquet("spark/output/out8/part-00000-56042b11-242d-4c2c-90de-4ea2e477a889-c000.snappy.parquet")
    //    resRDD1.show()

    /**
     * ========================================读写orc文件==========================================
     */
//    val studentDF: DataFrame = ss.read
//      .format("json")
//      .load("spark/data/students2.json")
//    //写入一个parquet格式的文件中
//    studentDF.write.orc("spark/output/out10")

//    val studentDF2: DataFrame = ss.read
//      .format("json")
//      .load("spark/data/students3.json")
//    //写入一个parquet格式的文件中
//    studentDF2.write.orc("spark/output/out11")
//    ss.read.orc("spark/output/out10/part-00000-6c395814-2894-44c8-ae02-64c1646dff5d-c000.snappy.orc").show()

    /**
     * ========================================读写jdbc文件==========================================
     */
    val consumerBehaviorDF: DataFrame = ss.read.format("jdbc")
      .option("url", "jdbc:mysql://master:3306/bigdata33?useUnicode=true&characterEncoding=utf-8&useSSL=false")
      .option("dbtable", "consumer_behavior")
      .option("user", "root")
      .option("password", "123456")
      .load()

    consumerBehaviorDF.show()



  }
}
